/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Copyright 2002-2003 by Mark Wolgemuth, All rights reserved.             */
/* Copyright 2002-2003 by Employease, Inc.  All rights reserved.           */
/*                                                                         */
/* Redistribution and use in source and binary forms, with or without      */
/* modification, are permitted provided that the following conditions      */
/* are met:                                                                */
/*                                                                         */
/*    1. Redistributions of source code must retain the above copyright    */
/* notice, this list of conditions and the following disclaimer.           */
/*    2. Redistributions in binary form must reproduce the above copyright */
/* notice, this list of conditions and the following disclaimer in the     */
/* documentation and/or other materials provided with the distribution.    */
/*                                                                         */
/* THIS SOFTWARE IS PROVIDED BY THE MOD_ATHENA PROJECT ``AS IS'' AND ANY   */
/* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE       */
/* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR      */
/* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FREEBSD PROJECT OR        */
/* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,   */
/* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,     */
/* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR      */
/* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY     */
/* OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT            */
/* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE   */
/* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.    */
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#include "athd.h"

/* private cache 
   this is how we determine last time we mailed 
   about an entity list
*/
static apr_hash_t *_athd_mail_hist = NULL;
static apr_pool_t *_athd_mail_hist_pool = NULL;
static apr_pool_t *_athd_mail_hist_subpool = NULL;
static pthread_mutex_t *_athd_mail_hist_mutex = NULL;

/* this base thread is private */
static void * _athd_job_thread(athd_job *job, 
			       int (*requester)(athd_job *),
			       int (*handler)(athd_job *, athd_request *));
/* these are the handlers for each type, also private */
static int _athd_job_thread_donothing_requester(athd_job *job);
static int _athd_job_thread_monitor_requester(athd_job *job);
static int _athd_job_thread_mirror_requester(athd_job *job);
static int _athd_job_thread_agent_requester(athd_job *job);
static int _athd_job_thread_donothing_handler(athd_job *j, athd_request *r);
static int _athd_job_thread_monitor_handler(athd_job *j, athd_request *r);
static int _athd_job_thread_mirror_handler(athd_job *j, athd_request *r);
static int _athd_job_thread_agent_handler(athd_job *j, athd_request *r);
static void _athd_job_thread_cleanup(void *ctx);
/* private mail history cache init and expire */
static void   _athd_job_mail_hist_init(apr_pool_t *pool);
static void   _athd_job_mail_hist_expire();
static int    _athd_job_mail_resend(athd_job *job, const char *key);
/* sig handler */
static int    _athd_job_thread_sig_start();
/* util obj for cache */
typedef struct athd_mail_hist athd_mail_hist;
struct athd_mail_hist {
    apr_time_t next_send;
};
static athd_mail_hist * athd_mail_hist_alloc();
static int    athd_mail_hist_lock();
static int    athd_mail_hist_unlock();
/* end private protos */

/* these four are the public requests */
void *
athd_job_thread_monitor(void *ctx)
{
    athd_job *job = ctx;
    ASSERT(job);
    _athd_job_thread(job, _athd_job_thread_monitor_requester,
		    _athd_job_thread_monitor_handler);
}

void *
athd_job_thread_mirror(void *ctx)
{
    athd_job *job = ctx;
    ASSERT(job);
    _athd_job_thread(job, _athd_job_thread_mirror_requester,
		    _athd_job_thread_mirror_handler);
}

void *
athd_job_thread_agent(void *ctx)
{
    athd_job *job = ctx;
    ASSERT(job);
    _athd_job_thread(job, _athd_job_thread_agent_requester,
		    _athd_job_thread_agent_handler);
}


void
athd_job_thread_shared_setup(apr_pool_t *pool)
{
    _athd_job_mail_hist_init(pool);
    athd_sig_handler_register(_athd_job_thread_sig_start, 
			      "Job Threads: mail history expire", ATHD_START);
}

/* everything below is private */

/* all work inside this thread should have a bounded CPU cost and no IO */
/* this thread needs to run on a reliable schedule */
static void *
_athd_job_thread(athd_job *job, 
		int (*requester)(athd_job *),
		int (*handler)(athd_job *, athd_request *))
{
    apr_time_t time_nextrun = 0;
    int laststate, lasttype;
    struct sigaction action = { NULL } ;
    ASSERT(job);

    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &laststate);
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &lasttype);
    pthread_cleanup_push((void *)_athd_job_thread_cleanup, (void *)job);

    action.sa_handler = athd_connection_reset;
    sigaction(SIGPIPE, &action, 0);

    athd_log(ATHD_INFO, "Job Thread: [%s] Enter", job->name);    
    /* iteration logic here */
    for (;(
	   (!job->dienow)	 
	   && ((job->iter_lives < 0) 
	       || (job->iter_count < job->iter_lives))
	   );
	 )
      {
	  int rv = APR_SUCCESS;
	  ath_node *listctx;
	  apr_time_t time_now = apr_time_now();
	  int response_throttle = 10;
	  athd_request *req = NULL;
	  const char *errmsg = "(null)";

	  pthread_testcancel();
	  /* check for responses that need to be handled first */
	  while ((!job->dienow) && (response_throttle > 1) 
		 && ((req = athd_job_shift_response(job)) != NULL)) 
	    {
		/* parsing here */
		athd_log(ATHD_DEVEL, "Job Thread: [%s] Handling [%s]",
			 job->name, req->url);
		rv = handler(job, req);
		if (rv) errmsg = "Response: ";
		/* done, remove it */
		athd_req_destroy(&req);
		response_throttle--;
	    }

	  /* next check for engine and request to send to it */
	  if ((!job->dienow) && (time_nextrun <= time_now))
	    {
		job->iter_count++;
		/* frequency is in job->freq seconds */
		time_nextrun = time_now + (apr_time_from_sec(job->freq));
		athd_log(ATHD_DEBUG, "Job Thread: [%s] Building requests",
			 job->name);
		rv = requester(job);
		if (rv) errmsg = "Request: ";
	    }
	  
	  if (ATHD_STATUS_IS_FATAL(rv))
	    {
		athd_log_err(ATHD_CRIT, rv, "Job Thread: [%s] Fatal %s",
			     job->name, errmsg);
		job->dienow = 1;
	    }
	  if (!job->dienow)
	      apr_sleep(XTH_SEC(10)); /* 1/10 sec pulse */
      }
    pthread_cleanup_pop(1);
}

static int
_athd_job_thread_donothing_requester(athd_job *job)
{
    /* chillin` */
    return APR_SUCCESS;
}

static int
_athd_job_thread_monitor_requester(athd_job *job)
{
    int rv = APR_SUCCESS;
    athd_request *req;
    ASSERT(job);
    req = athd_req_create_for_job(job, NULL, ATHD_REQ_SERIAL_PHYS, 
				  (const void *)ATHD_PHYS_ALL);
    if (req)
	rv = athd_engine_push_req(req);
    else
	rv = APR_EGENERAL;
    return rv;
}

/* mirror requests go to job's engine (serialized status reqs)
   but mirror handler generates reqs to secondaries
   handler then must get it again and handle secondary response
*/
static int
_athd_job_thread_mirror_requester(athd_job *job)
{
    int rv = APR_SUCCESS;
    athd_request *req = NULL;
    ASSERT(job);
    req = athd_req_create_for_job(job, NULL, ATHD_REQ_SERIAL_PHYS, 
				  (const void *)ATHD_PHYS_ALL);
    if (req)
	rv = athd_engine_push_req(req);
    else
	rv = APR_EGENERAL;
    if (!rv)
      {
	  req = athd_req_create_for_job(job, NULL, ATHD_REQ_SERIAL_FARM, 
					(const void *)ATHD_FARM_ALL);
	  if (req)
	      rv = athd_engine_push_req(req);
	  else 
	      rv = APR_EGENERAL;
      }
    if (rv)
	athd_req_destroy(&req);
    return rv;
}

static int
_athd_job_thread_agent_requester(athd_job *job)
{
    int rv = APR_SUCCESS;
    athd_request *req;
    /* get stats as appropriate */
    req = athd_req_create_for_job(job, NULL, ATHD_REQ_HELLO, NULL);
    if (req)
	rv = athd_engine_push_req(req);
    else
	rv = APR_EGENERAL;
    return rv;
}

static int
_athd_job_thread_donothing_handler(athd_job *job, athd_request *req)
{
    /* chillin` */
    return APR_SUCCESS;
}

static int
_athd_job_thread_monitor_handler(athd_job *job, athd_request *req)
{
    athd_mail *mail = NULL;
    athd_log(ATHD_DEVEL, "Job Thread Monitor Handler: "
	     "[%s] Request: [%s] Response: [%s]",
	     job->name, req->url, req->response_txt);
    /* error is always going to be at front of str */
    if ((ath_str_is_error(req->response_txt))
	|| (ath_str_is_warn(req->response_txt)))
	athd_log(ATHD_WARN, 
		 "Job: [%s] Engine: [%s] Remote Msg: [%s]",
		 job->name, req->engine->name, req->response_txt);
    else
      {
	  char *c;
	  char *resp_pos = req->response_txt;
	  char *line, *intern;
	  char *sep = "\n";
	  /* built in req pool then copied to mail */
	  char *downers = NULL;

	  line = apr_strtok(resp_pos, sep, &intern);
	  while (line)
	    {
		if (req->type == ATHD_REQ_SERIAL_PHYS)
		  {
		      ath_phys *phys = 
			ath_phys_clone_from_str(req->pool, line);
		      if (!phys)
			  athd_log(ATHD_DEBUG, 
				   "Job: [%s] Engine: [%s] " 
				   "Failed to deserialize phys str: [%s]",
				   job->name, req->engine->name, line);
		      /* warn only if online and down */
		      else if (((!phys->up) || (!phys->in))
				&& (phys->on))
			{
			    if (!downers)
				downers = 
				  apr_pstrcat(req->pool, "Hosts down: ", 
					      phys->host, NULL);
			    else
				downers = 
				  apr_pstrcat(req->pool, downers,
					      ", ", phys->host,	NULL);
			    athd_log(ATHD_WARN, "Job: [%s] Engine: [%s] " 
				     "Phys host down: [%s]",
				     job->name, req->engine->name, phys->host);
			}
		      line = apr_strtok(NULL, sep, &intern);
		  }
	    }
	  /* the key is the downers list, which should always
	     list in the same order-- 
	     so it will not resend if this list hasn't changed
	     in the resend throttle window
	     see athd_mail_resend
	  */
	  if ((downers) && (job->mail)
	      && (_athd_job_mail_resend(job, downers)))
	    {
		/* must tie this to job pool... it may out live req */
		mail = athd_mail_create(job->pool);
		mail->subject = apr_pstrcat(mail->pool, "Host down on ", 
					    job->engine->name, NULL);
		mail->body = apr_pstrdup(mail->pool, downers);
		athd_mail_build_for_job(job, mail);
		athd_mail_queue_add(mail);
	    }
      }
    return APR_SUCCESS;
}

/* mirror requests go to job's engine 
   but mirror handler generates reqs to secondary engines
   handler then must get it again, handle secondary response
*/
static int
_athd_job_thread_mirror_handler(athd_job *job, athd_request *req)
{
    int rv = APR_SUCCESS;
    ASSERT(job->engine);
    athd_log(ATHD_DEVEL, "Job Thread Mirror Handler: " 
	     "[%s] Request: [%s] Response: [%s]",
	     job->name, req->url, req->response_txt);
    /* handle error first, this catches both pri and sec errors */
    if ((ath_str_is_error(req->response_txt))
	||(ath_str_is_warn(req->response_txt)))
	athd_log(ATHD_WARN, 
		 "Job: [%s] Engine: [%s] Remote Msg: [%s]",
		 job->name, req->engine->name, req->response_txt);
    /* generate mirroring phys and farm reqs from primary */
    else if ((req->type == ATHD_REQ_SERIAL_PHYS)
	|| (req->type == ATHD_REQ_SERIAL_FARM))
      {
	  char *c;
	  char *resp_pos = req->response_txt;
	  char *line, *intern;
	  char *sep = "\n";
		
	  line = apr_strtok(resp_pos, sep, &intern);
	  /* each line is a serialized farm or phys
	     no need to parse it, just forward it to secondary
	  */
	  while (line) 
	    {
		ath_node *listctx;
		ath_list_rewind_r(job->engine->secondaries, &listctx);
		while (ath_list_hasnext_r(job->engine->secondaries, &listctx))
		  {
		      athd_engine *engsec = (athd_engine *)
			ath_list_getnext_r(job->engine->secondaries, &listctx);
		      athd_request *reqmir = NULL;
		      if (req->type == ATHD_REQ_SERIAL_PHYS)
			  reqmir =
			    athd_req_create_for_job(job, engsec, 
						    ATHD_REQ_MIRROR_PHYS,
						    (const void *)line);
		      else if (req->type == ATHD_REQ_SERIAL_FARM)
			  reqmir =
			    athd_req_create_for_job(job, engsec, 
						    ATHD_REQ_MIRROR_FARM,
						    (const void *)line);
		      if (reqmir)
			  rv = athd_engine_push_req(reqmir);
		      else
			  rv = APR_EGENERAL;
		  }
		line = apr_strtok(NULL, sep, &intern);
	    }
      }
    /* nothing to be done with secondary's responses */
    return rv;
}

static int
_athd_job_thread_agent_handler(athd_job *job, athd_request *req)
{
    athd_log(ATHD_DEVEL, "Job Thread Agent Handler: "
	     "[%s] Request: [%s] Response: [%s]",
	     job->name, req->url, req->response_txt);
    /* do nothing but note errors in response */
    if ((ath_str_is_error(req->response_txt))
	||(ath_str_is_warn(req->response_txt)))
	athd_log(ATHD_WARN, 
		 "Job: [%s] Engine: [%s] Remote Msg: [%s]",
		 job->name, req->engine->name, req->response_txt);
    return APR_SUCCESS;
}


static void 
_athd_job_mail_hist_expire()
{
    /* we could clear, but to free to os we destroy */
    /* since this will be rarely called, it's no perf issue */
    athd_mail_hist_lock();
    apr_pool_destroy(_athd_mail_hist_subpool);
    athd_mail_hist_unlock();
    _athd_job_mail_hist_init(NULL);
}

static void 
_athd_job_mail_hist_init(apr_pool_t *pool)
{
    if (!pool)
	ASSERT(_athd_mail_hist_pool);
    else
      {
	  ASSERT(!_athd_mail_hist_pool);
	  apr_pool_create(&_athd_mail_hist_pool, pool);
      }
    if (!_athd_mail_hist_mutex)
      {	
	  _athd_mail_hist_mutex = 
	    apr_pcalloc(_athd_mail_hist_pool, sizeof(pthread_mutex_t));
	  apr_pool_cleanup_register(_athd_mail_hist_pool, 
				    _athd_mail_hist_mutex,
				    pthread_mutex_cleanup, 
				    apr_pool_cleanup_null);
	  pthread_mutex_init(_athd_mail_hist_mutex, NULL);
      }
    athd_mail_hist_lock();
    apr_pool_create(&_athd_mail_hist_subpool,
		    _athd_mail_hist_pool);
    _athd_mail_hist = apr_hash_make(_athd_mail_hist_subpool);
    athd_mail_hist_unlock();
    athd_log(ATHD_DEBUG, "Job Threads: Mail history cache initialized.");
}

/* becareful with key collisions if adding new handlers that use this.
   only monitor uses it right now, and it uses list of down hosts
   as key.
*/
static int
_athd_job_mail_resend(athd_job *job, const char *key)
{
    int rv = 1;
    ASSERT(job);
    ASSERT(key);
    if (job->mail_resend_throttle > 0)
      {
	  apr_time_t time;
	  athd_mail_hist *hist = NULL;

	  athd_mail_hist_lock();
	  time = apr_time_now();
	  hist = (athd_mail_hist *)
	    apr_hash_get(_athd_mail_hist, key,
			 APR_HASH_KEY_STRING);
	  if (!hist)
	    {
		char *mykey = apr_pstrdup(_athd_mail_hist_subpool, key);
		hist = athd_mail_hist_alloc();
		apr_hash_set(_athd_mail_hist, mykey,
			     APR_HASH_KEY_STRING, hist);
	    }		 
	  if (time < hist->next_send)
	      rv = 0;
	  else 
	    {
		/* this would keep as close as possible to schedule,
		   but it also risks infinitely growing queue 
		   (if next_send < time to often )
		*/
		if (hist->next_send == 0)
		    hist->next_send = time;
		hist->next_send += job->mail_resend_throttle;
	      /* this is alternative code */
	      /* hist->next_send = time + job->mail_resend_throttle; */
	    }
	  athd_mail_hist_unlock();
      }
    return rv;
}

static athd_mail_hist *
athd_mail_hist_alloc()
{
    athd_mail_hist *hist = NULL;
    ASSERT(_athd_mail_hist_subpool);
    hist = apr_pcalloc(_athd_mail_hist_subpool, 
		       sizeof(athd_mail_hist));
    hist->next_send = 0;
    return hist;
}

static int
athd_mail_hist_lock()
{
    return pthread_mutex_lock(_athd_mail_hist_mutex);
}

static int
athd_mail_hist_unlock()
{
    return pthread_mutex_unlock(_athd_mail_hist_mutex);
}

static int
_athd_job_thread_sig_start(int sig)
{
    _athd_job_mail_hist_expire();
    return APR_SUCCESS;
}

static void
_athd_job_thread_cleanup(void *ctx)
{
    athd_job *job = (athd_job *)ctx;
    athd_log(ATHD_INFO, "Job Thread: [%s] Cleanup/Exit", job->name); 
}
